package main

import (
	"bufio"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"rpc/msg"
	"sync"
	"sync/atomic"
	"time"
)

func main() {
	server := NewServer("0.0.0.0:8080")
	server.Start()
	select {}
}

var (
	serverPool = &sync.Pool{
		New: func() interface{} {
			return make([]byte, 0, 1024)
		}}

	server8bPool = &sync.Pool{
		New: func() interface{} {
			return make([]byte, 8)
		}}
)

const serverBatchSize = 2 * 1024 * 1024

type Server struct {
	Addr  string
	count atomic.Int32
}

func NewServer(addr string) *Server {
	return &Server{
		Addr: addr,
	}
}

type RpcConn struct {
	conn         net.Conn
	waitSendChan chan *msg.Msg
	lock         *sync.Mutex
}

func NewRpcConn(conn net.Conn) *RpcConn {
	return &RpcConn{
		conn:         conn,
		waitSendChan: make(chan *msg.Msg, 10240),
		lock:         new(sync.Mutex),
	}
}

func (server *Server) Start() {
	listener, err := net.Listen("tcp4", server.Addr)
	if err != nil {
		panic(err)
	}

	fmt.Println("Server start...")
	for {
		conn, err := listener.Accept()
		server.count.Add(1)
		fmt.Printf("recv %d client \n", server.count.Load())
		if err != nil {
			panic(err)
		}
		rpcConn := NewRpcConn(conn)
		go server.handleSend(rpcConn)
		go server.handleRead(rpcConn)
	}
}

func (server *Server) handleSend(rpcConn *RpcConn) {
	ticker := time.NewTicker(10 * time.Millisecond)
	packet := serverPool.Get().([]byte)
	packetSize := 0
	for {
		select {
		case ms := <-rpcConn.waitSendChan:
			d := ms.Encode()
			packetSize = packetSize + len(d)
			packet = append(packet, d...)
			if packetSize > serverBatchSize {
				p := msg.NewPacket(&packet)
				rpcConn.conn.Write(p.Encode())
				packet = make([]byte, 0, 1024)
				serverPool.Put(packet)
				packetSize = 0
			}

		case <-ticker.C:
			if len(packet) != 0 {
				p := msg.NewPacket(&packet)
				rpcConn.conn.Write(p.Encode())
				packet = make([]byte, 0, 1024)
				serverPool.Put(packet)
				packetSize = 0
			}
		}
	}
}

func (server *Server) handleRead(rpcConn *RpcConn) {
	conn := rpcConn.conn
	defer func() {
		if conn != nil {
			fmt.Println(conn.RemoteAddr().String() + " closed")
			conn.Close()
		}
	}()
	reader := bufio.NewReader(conn)
	for {
		header := server8bPool.Get().([]byte)
		_, err := io.ReadFull(reader, header)
		if err != nil {
			if err == io.EOF {
				return
			}
			return
		}
		l := binary.BigEndian.Uint64(header[:8])
		server8bPool.Put(header)

		data := make([]byte, l)
		_, err = io.ReadFull(reader, data)
		if err != nil {
			if err == io.EOF {
				return
			}
			return
		}
		p := &msg.Packet{
			Header: l,
			Batch:  &data,
		}
		go handlePacket(p, rpcConn)
	}
}

func handlePacket(packet *msg.Packet, rpcConn *RpcConn) {
	batch := *packet.Batch
	i := 0
	for i < len(batch) {
		l := binary.BigEndian.Uint64(batch[i : 8+i])
		seqNo := binary.BigEndian.Uint64(batch[8+i : 16+i])
		endIdx := 16 + i + int(l) - 8
		req := &msg.Msg{
			Length: l,
			SeqNo:  seqNo,
			Data:   batch[16+i : endIdx],
		}
		i = endIdx
		asyncHandleMsg(req, rpcConn)
	}
}

func asyncHandleMsg(ms *msg.Msg, rpcConn *RpcConn) {

	//data := ms.Data
	//hash32 := crc32.New(crc32.IEEETable)
	//hash32.Write(data)
	//sum32 := hash32.Sum32()
	//byteBuf := make([]byte, 8)
	//binary.BigEndian.PutUint32(byteBuf,sum32)
	//
	//ms.Data = byteBuf
	//ms.Length = uint64(8 + len(byteBuf))
	rpcConn.waitSendChan <- ms
}
